<?php

use Illuminate\Database\Capsule\Manager as DB;

class JobController extends CoreController
{

    public function init()
    {
        if (!$this->isCli()) {
            throw new Exception("Go to cli please!", '303');
        }
        parent::init();
    }

    public function stopAction(){
        configModel::where('type', 11)->where('key', 'queueRestart')->update(['value'=>time()]);
        Log::out("任务队列退出", 'job');
        json(['ret'=>0, 'msg'=>'任务队列退出']);
    }

    /**
     * Execute the console command.
     * @param Input  $input
     * @param Output $output
     * @return int|null|void
     */
    public function workAction()
    {
        $queue = $this->get('queue', 'shop');
        $delay = $this->get('delay', 0);
        $memory= $this->get('memory', 128);
        $sleep = $this->get('sleep', 5);
        $tries = $this->get('tries', 0);

        if ($this->get('daemon', 0)) {
            $this->daemon($queue, $delay, $memory, $sleep, $tries);
        } else {
            $response = $this->pop($queue, $delay, $sleep, $tries);
            $this->output($response);
        }
    }

    /**
     * 启动一个守护进程执行任务.
     *
     * @param  string $queue
     * @param  int    $delay
     * @param  int    $memory
     * @param  int    $sleep
     * @param  int    $maxTries
     * @return array
     */
    protected function daemon($queue = null, $delay = 0, $memory = 128, $sleep = 3, $maxTries = 0)
    {
        $lastRestart = $this->getTimestampOfLastQueueRestart();

        while (true) {
            $this->runNextJobForDaemon(
                $queue, $delay, $sleep, $maxTries
            );

            if ( $this->memoryExceeded($memory) ) {
                Hook::emit('worker_memory_exceeded', $queue);
                $this->stop();
            }

            if ( $this->queueShouldRestart($lastRestart) ) {
                Hook::emit('worker_queue_restart', $queue);
                $this->stop();
            }
        }
    }

    /**
     * 检查内存是否超出
     * @param  int $memoryLimit
     * @return bool
     */
    protected function memoryExceeded($memoryLimit)
    {
        return (memory_get_usage() / 1024 / 1024) >= $memoryLimit;
    }


    /**
     * 获取上次重启守护进程的时间
     *
     * @return int|null
     */
    protected function getTimestampOfLastQueueRestart()
    {
        return configModel::where('type', 11)->where('key', 'queueRestart')->value('value');
    }

    /**
     * 检查是否要重启守护进程
     *
     * @param  int|null $lastRestart
     * @return bool
     */
    protected function queueShouldRestart($lastRestart)
    {
        return $this->getTimestampOfLastQueueRestart() != $lastRestart;
    }

    /**
     * 以守护进程的方式执行下个任务.
     *
     * @param  string $queue
     * @param  int    $delay
     * @param  int    $sleep
     * @param  int    $maxTries
     * @return void
     */
    protected function runNextJobForDaemon($queue, $delay, $sleep, $maxTries)
    {
        try {
            $response = $this->pop($queue, $delay, $sleep, $maxTries);

            $this->output($response);
        } catch (Exception $e) {
            Log::out($e->getMessage(), 'job');
        } catch (Throwable $e) {
            Log::out($e->getMessage(), 'job');
        }
    }


    protected function output($response)
    {
        if (!is_null($response['job'])) {
            $job = $response['job'];
            if ($response['failed']) {
                Log::out('Failed: ' . $job->getName(), 'job');
            } else {
                Log::out('Processed: ' . $job->getName(), 'job');
            }
        }
    }

    /**
     * 执行下个任务
     * @param  string $queue
     * @param  int    $delay
     * @param  int    $sleep
     * @param  int    $maxTries
     * @return array
     */
    public function pop($queue = null, $delay = 0, $sleep = 3, $maxTries = 0)
    {
        $job = $this->getNextJob($queue);

        if (!is_null($job)) {
            #Hook::emit('worker_before_process', '[' . $queue . ']' . $job->getName());
            return $this->process($job, $maxTries, $delay);
        }

        $this->sleep($sleep);

        return ['job' => null, 'failed' => false];
    }

    /**
     * 获取下个任务
     * @param  string $queue
     * @return QueueJob
     */
    protected function getNextJob($queue)
    {
        if (is_null($queue)) {
            return Queue::pop();
        }

        foreach (explode(',', $queue) as $queue) {
            if (!is_null($job = Queue::pop($queue))) {
                return $job;
            }
        }
    }

    /**
     * Process a given job from the queue.
     * @param  QueueJob $job
     * @param  int             $maxTries
     * @param  int             $delay
     * @return array
     * @throws Exception
     */
    public function process(QueueJob $job, $maxTries = 0, $delay = 0)
    {
        if ($maxTries > 0 && $job->attempts() > $maxTries) {
            return $this->logFailedJob($job);
        }

        try {
            $job->fire();

            return ['job' => $job, 'failed' => false];
        } catch (Exception $e) {
            if (!$job->isDeleted()) {
                $job->release($delay);
            }

            throw $e;
        }
    }

    /**
     * Log a failed job into storage.
     * @param  QueueJob $job
     * @return array
     */
    protected function logFailedJob(QueueJob $job)
    {
        if (!$job->isDeleted()) {
            try {
                $job->delete();
                $job->failed();
            } finally {
                Hook::emit('worker_queue_failed', $job);
            }
        }

        return ['job' => $job, 'failed' => true];
    }

    /**
     * Sleep the script for a given number of seconds.
     * @param  int $seconds
     * @return void
     */
    public function sleep($seconds)
    {
        sleep($seconds);
    }

    /**
     * 停止执行任务的守护进程.
     * @return void
     */
    public function stop()
    {
        exit();
    }

}
